#ifndef __M_CONSUMER_H__
#define __M_CONSUMER_H__
#include "../mqcommon/helper.hpp"
#include "../mqcommon/log.hpp"
#include "../mqcommon/mq_meg.pb.h"

#include <iostream>

namespace pprmq
{
    // 回调函数的作用:解耦消息接收与消息处理逻辑
    using ConsumerCallback = std::function<void(const std::string, const BasicProperties *bp, const std::string)>;
    class Consumer
    {
    public:
        Consumer() {}
        Consumer(std::string ctag, std::string queue_qname, bool auto_ack, ConsumerCallback cb)
            : _tag(ctag), _qname(queue_qname), _auto_ack(auto_ack), _callback(cb)
        {}

    public:
        using ptr = std::shared_ptr<Consumer>;
        std::string _tag;   // 消费者的标识
        std::string _qname; // 消费者订阅的队列
        bool _auto_ack;     // 自动确认状态
        ConsumerCallback _callback;
    };

    // 以队列为单元的消费者管理结构
    class QueueConsumer
    {
    public:
        using ptr = std::shared_ptr<QueueConsumer>;
        QueueConsumer(const std::string &qname) : _qname(qname), _rr_seq(0) {}
        // 队列新增消费者
        Consumer::ptr create(const std::string &ctag, const std::string &queue_name, bool ack_flag, const ConsumerCallback &cb)
        {
            // 1. 加锁
            std::unique_lock<std::mutex> lock(_mutex);
            // 2. 判断消费者是否重复
            for (auto &consumer : _consumers)
            {
                if (consumer->_tag == ctag)
                {
                    return Consumer::ptr();
                }
            }
            // 3. 没有重复则新增--构造对象
            Consumer::ptr consumer = std::make_shared<Consumer>(ctag, queue_name, ack_flag, cb);
            // 4. 添加管理后返回对象
            _consumers.push_back(consumer);
            return consumer;
        }
        // 队列移除消费者
        void remove(const std::string &ctag)
        {
            // 1. 加锁
            std::unique_lock<std::mutex> lock(_mutex);
            // 2. 看找不找得到消费者
            for (auto it = _consumers.begin(); it != _consumers.end(); it++)
            {
                if ((*it)->_tag == ctag)
                {
                    // 找到了
                    _consumers.erase(it);
                    return;
                }
            }
            return;
        }
        // 队列获取消费者：RR轮转获取
        Consumer::ptr choose()
        {
            // 1. 加锁
            std::unique_lock<std::mutex> lock(_mutex);
            if (_consumers.size() == 0)
                return Consumer::ptr();
            int idx = _rr_seq % _consumers.size();
            _rr_seq++;              // 序号自增，下次轮询到下一个
            return _consumers[idx]; // 返回选中的消费者
        }
        // 是否为空
        bool empty()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _consumers.empty();
        }
        // 判断指定消费者是否存在
        bool exists(const std::string &ctag)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 2. 看找不找得到消费者
            for (auto it = _consumers.begin(); it != _consumers.end(); it++)
            {
                if ((*it)->_tag == ctag)
                {
                    // 找到了
                    return true;
                }
            }
            return false;
        }
        // 清理所有消费者
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _consumers.clear();
            _rr_seq = 0;
        }

    private:
        std::mutex _mutex;
        std::string _qname;
        uint64_t _rr_seq; // 轮转序号
        std::vector<Consumer::ptr> _consumers;
    };

    class ConsumerManager
    {
    public:
        using ptr = std::shared_ptr<ConsumerManager>;
        ConsumerManager() {}
        void initQueueConsumer(const std::string &qname)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _qconsumers.find(qname);
            if (it != _qconsumers.end())
                return;
            // 没找到
            auto qcp = std::make_shared<QueueConsumer>(qname);
            _qconsumers.insert(std::make_pair(qname, qcp));
        }
        void destroyQueueConsumer(const std::string &qname)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _qconsumers.erase(qname);
        }
        Consumer::ptr create(const std::string &ctag, const std::string &queue_name, bool ack_flag, const ConsumerCallback &cb)
        {
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _qconsumers.find(queue_name);
                if (it == _qconsumers.end())
                {
                    // 没找到
                    return Consumer::ptr();
                }
                qcp = it->second;
            }
            return qcp->create(ctag, queue_name, ack_flag, cb);
        }

        void remove(const std::string &ctag, const std::string &queue_name)
        {
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _qconsumers.find(queue_name);
                if(it == _qconsumers.end())
                {
                    return ;
                }
                qcp = it->second;
            }
            return qcp->remove(ctag);
        }
        Consumer::ptr choose(const std::string &queue_name)
        {
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _qconsumers.find(queue_name);
                if(it == _qconsumers.end())
                {
                    return Consumer::ptr();
                }
                qcp = it->second;
            }
            return qcp->choose();
        }

        bool empty(const std::string &queue_name)
        {
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _qconsumers.find(queue_name);
                if(it == _qconsumers.end())
                {
                    return false;
                }
                qcp = it->second;
            }
            return qcp->empty(); 
        }

        bool exists(const std::string &ctag, const std::string &queue_name)
        {
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _qconsumers.find(queue_name);
                if(it == _qconsumers.end())
                {
                    return false;
                }
                qcp = it->second;
            }
            return qcp->exists(ctag);
        }

        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _qconsumers.clear();
        }

    private:
        std::mutex _mutex;
        std::unordered_map<std::string, QueueConsumer::ptr> _qconsumers;
    };
}

#endif